package com.hmeter.hadoop.mapreduce.jobtracker;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapred.Counters.Group;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobID;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.Counters;

public class JobCountersMetric {

	public JobCountersMetric() {};
	
	protected static JobClient prepareJobClient(String jobTracker) throws IOException
	{
		Configuration conf = new Configuration();
		conf.set("mapred.job.tracker", jobTracker);
		JobConf jobConf = new JobConf(conf);
		JobClient client = new JobClient(jobConf);
		return client;
	}
	
	protected static JobID convertToJobId(String jobId)
	{
		String id = jobId.replace("job_", "");
		return new JobID(id.split("_")[0],Integer.valueOf(id.split("_")[1]));
	}
	
	public String getJobCountersMetric(String jobTracker, String jobId, String groupName) throws IOException {
		
		StringBuilder jobCountersResult = new StringBuilder();
		
		JobClient client = JobCountersMetric.prepareJobClient(jobTracker);
    	RunningJob job = client.getJob(convertToJobId(jobId));
    	
    	Counters counters = job.getCounters();	
		
		jobCountersResult.append("<jobmetric jobid='" + jobId + "' jobname='"+job.getJobName()+"'>\n");
		jobCountersResult.append(" <counters>\n");
		
		if (!groupName.equals(""))
		{
			for (Group counterGroup : counters)
			{
				if (counterGroup.getDisplayName().equals(groupName))
				{
					for (Counter counter : counterGroup)
					{					
						jobCountersResult.append("  <counter name='" + counter.getDisplayName() + "' value='" + counter.getValue() + "'>\n");
					}
					break;
				}	
			}
		}
		else
		{
			for (Group counterGroup : counters)
			{
				for (Counter counter : counterGroup)
				{					
					jobCountersResult.append("  <counter name='" + counter.getDisplayName() + "' value='" + counter.getValue() + "'>\n");
				}	
			}
		}
		
		jobCountersResult.append(" </counters>\n");
		jobCountersResult.append("<jobmetric>\n");
		
		return jobCountersResult.toString();
	}
	
	public String getJobStatusMetric(String jobTracker, String jobId) throws IOException
	{
		StringBuilder jobStatusResult = new StringBuilder();
		
		JobClient client = prepareJobClient(jobTracker);
		RunningJob job = client.getJob(convertToJobId(jobId));

		double setupProgress = job.setupProgress() * 100;
		double mapProgress = job.mapProgress() * 100;
		double reduceProgress = job.reduceProgress() * 100;
		double cleanupProgress = job.cleanupProgress() * 100;
		String setupProgressPercentage = Double.toString(setupProgress) + "%";
		String mapProgressPercentage = Double.toString(mapProgress) + "%";
		String reduceProgressPercentage = Double.toString(reduceProgress) + "%";
		String cleanupProgressPercentage = Double.toString(cleanupProgress) + "%";

		jobStatusResult.append("<jobmetric jobid='"+ jobId +"'" + " jobname='"+ job.getJobName() +"'>\n");
		jobStatusResult.append("  <setupProgress>"+setupProgressPercentage+"</setupProgress>\n");
		jobStatusResult.append("  <mapProgress>"+mapProgressPercentage+"</mapProgress>\n");
		jobStatusResult.append("  <reduceProgress>"+reduceProgressPercentage+"</reduceProgress>\n");
		jobStatusResult.append("  <cleanupProgress>"+cleanupProgressPercentage+"</cleanupProgress>\n");
		jobStatusResult.append("  <complete>"+job.isComplete()+"</complete>\n");
		jobStatusResult.append("  <successful>"+job.isSuccessful()+"</successful>\n");
		jobStatusResult.append("  <url>"+job.getTrackingURL()+"</url>\n");
		jobStatusResult.append("</jobmetric>");
		
		return jobStatusResult.toString();
	}
}
